package com.lgj.rocketmq.config;

import com.lgj.rocketmq.code.ConsumerInitFactory;
import com.lgj.rocketmq.code.ConsumerRocketMQListenerContainer;
import com.lgj.rocketmq.code.MQPushConsumerFactory;
import com.lgj.rocketmq.code.ProducerTransactionFactory;
import com.lgj.rocketmq.product.ProducerTransactionListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.spring.autoconfigure.RocketMQAutoConfiguration;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Slf4j
@Configuration
@ConditionalOnMissingBean({RocketMQAutoConfiguration.class})
public class RocketAutoConfiguration {
    @Autowired
    private ProducerProperties producerProperties;

    @Autowired
    private ConsumerProperties consumerProperties;

    @Value("${spring.application.name}")
    private String applicationName;

    @Autowired
    private ProducerTransactionFactory productTransactionFactory;

    @Bean
    public ConsumerInitFactory consumerInitFactory(){
        return new ConsumerInitFactory();
    }

    @Bean
    public MQPushConsumerFactory mqPushConsumerFactory(){
        return new MQPushConsumerFactory(consumerProperties,consumerInitFactory());
    }

    @Bean
    @ConditionalOnMissingBean(ConsumerRocketMQListenerContainer.class)
    public ConsumerRocketMQListenerContainer rocketMessageListenerContainer() {
        ConsumerRocketMQListenerContainer container =
                new ConsumerRocketMQListenerContainer(mqPushConsumerFactory());
        return container;
    }


    @Bean
    public TransactionMQProducer transactionMQProducer() throws MQClientException {
        TransactionMQProducer defaultMQProducer = new TransactionMQProducer();
        defaultMQProducer.setProducerGroup(applicationName);
        defaultMQProducer.setNamesrvAddr(producerProperties.getNameServerAddress());
        if (producerProperties.getMaxRetry() != null){
            defaultMQProducer.setRetryTimesWhenSendFailed(producerProperties.getMaxRetry());
        }
        if (producerProperties.getRetryAnotherBroker() != null){
            defaultMQProducer.setRetryAnotherBrokerWhenNotStoreOK(producerProperties.getRetryAnotherBroker());
        }
        defaultMQProducer.setTransactionListener(new ProducerTransactionListener(productTransactionFactory));
        defaultMQProducer.start();
        log.info("生产者【{}】启动成功！", applicationName);
        return defaultMQProducer;
    }
}
